package io.plus.engine.pulse.task;

import io.plus.engine.pulse.AsynchronousSocketChannelContext;
import io.plus.engine.pulse.Configuration;
import io.plus.engine.pulse.PacketHandlerMode;
import io.plus.engine.pulse.PacketMetadata;
import io.plus.engine.pulse.task.queue.Queue;
import io.plus.engine.pulse.task.queue.QueueImpl;
import io.plus.engine.utils.lock.ReadWriteLockMap;
import io.plus.engine.utils.lock_capable_task.AbstractQueueTask;
import lombok.extern.slf4j.Slf4j;
import tool.plus.Clock;

import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class HandlerTask extends AbstractQueueTask<PacketMetadata> {


    private AsynchronousSocketChannelContext asynchronousSocketChannelContext = null;
    private Configuration configuration = null;

    private AtomicLong synFailCount = new AtomicLong();
    private Queue<PacketMetadata> queue = null;

    public HandlerTask(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Executor executor) {
        super(executor);
        this.asynchronousSocketChannelContext = asynchronousSocketChannelContext;
        configuration = asynchronousSocketChannelContext.clientConfiguration;
        getQueue();
    }

    @Override
    public Queue<PacketMetadata> getQueue() {
        if (PacketHandlerMode.thread_pool == configuration.packetHandlerMode) {
            if (queue == null) {
                synchronized (this) {
                    if (queue == null) {
                        queue = new QueueImpl<PacketMetadata>();
                    }
                }
            }
            return queue;
        }
        return null;
    }

    @Override
    public void runTask() {
        PacketMetadata packetMetadata = null;
        while ((packetMetadata = queue.poll()) != null) {
            handler(packetMetadata);
        }
    }

    /**
     * 处理packet
     */
    public void handler(PacketMetadata packetMetadata) {
        long start = Clock.currentTimeMillis;
        try {
            Integer synchronismSerialNumber = packetMetadata.getSynchronismSerialNumber();
            if (synchronismSerialNumber != null && synchronismSerialNumber > 0) {
                ReadWriteLockMap<Integer, PacketMetadata> synchronismSerialNumberToPacketMap = configuration.getSynchronismSerialNumberToPacketMap();
                PacketMetadata remove = synchronismSerialNumberToPacketMap.remove(synchronismSerialNumber);
                if (remove != null) {
                    synchronized (remove) {
                        synchronismSerialNumberToPacketMap.put(synchronismSerialNumber, packetMetadata);
                        remove.notify();
                    }
                } else {
                    log.error("[{}]同步消息失败, synchronismSerialNumber is {}, 但是同步集合中没有对应key值", synFailCount.incrementAndGet(), synchronismSerialNumber);
                }
            } else {
                configuration.getHandler().handler(asynchronousSocketChannelContext, packetMetadata);
            }
        } catch (Throwable e) {
            log.error(packetMetadata.toString(), e);
        } finally {


        }
    }


}
